# -*- coding: utf-8 -*-
# @Time    : 2025/10/14
# @Author  : 楚江涛
# @File    : bgloop.py
# @Software: PyCharm
#
# 全局后台事件循环（BackgroundLoop）
# 目标：在“同步上下文”（Django/Ninja 普通视图、manage.py shell 等）里也能
#       把协程提交到一个常驻的 asyncio 事件循环中执行，避免任务卡在 PENDING。
#
# 用法：
#   from apps.workflow.infrastructure.bgloop import bg_loop
#   bg_loop.submit(coro)  # 返回 concurrent.futures.Future
#
# 说明：
# - 模块加载时即创建并启动一个后台事件循环线程（daemon）。
# - submit() 内部使用 asyncio.run_coroutine_threadsafe 安全跨线程提交协程。
# - stop() 仅在进程退出或测试场景使用；生产一般无需手动调用。

from __future__ import annotations

import asyncio
import threading
from typing import Awaitable, Any, Optional
from concurrent.futures import Future


class BackgroundLoop:
    """全局后台事件循环封装（线程安全、可复用）。"""

    def __init__(self) -> None:
        self._lock = threading.RLock()
        self._loop: Optional[asyncio.AbstractEventLoop] = None
        self._thread: Optional[threading.Thread] = None
        self._start_if_needed()

    # ---------------------------
    # 生命周期
    # ---------------------------

    def _start_if_needed(self) -> None:
        with self._lock:
            if self._loop and self._loop.is_running():
                return
            loop = asyncio.new_event_loop()
            t = threading.Thread(target=loop.run_forever, name="bg-asyncio-loop", daemon=True)
            t.start()
            self._loop = loop
            self._thread = t

    def stop(self) -> None:
        """停止后台事件循环（测试/退出时使用）。"""
        with self._lock:
            if not self._loop:
                return
            try:
                self._loop.call_soon_threadsafe(self._loop.stop)
            finally:
                self._loop = None
                self._thread = None

    # ---------------------------
    # 提交任务
    # ---------------------------

    def submit(self, coro: Awaitable[Any]) -> Future:
        """
        提交协程到后台事件循环执行，返回 concurrent.futures.Future。
        - 可在同步线程调用；
        - 执行中的异常会体现在返回的 Future 上（.result() 时抛）。
        """
        if not asyncio.iscoroutine(coro):
            raise TypeError("bg_loop.submit 需要一个协程对象（awaitable coroutine）。")
        self._start_if_needed()
        assert self._loop is not None
        return asyncio.run_coroutine_threadsafe(coro, self._loop)

    def call_soon(self, fn, *args, **kwargs) -> None:
        """把一个同步回调投递到后台 loop 的线程中尽快执行（非协程）。"""
        self._start_if_needed()
        assert self._loop is not None
        self._loop.call_soon_threadsafe(fn, *args, **kwargs)

    # ---------------------------
    # 辅助
    # ---------------------------

    def is_running(self) -> bool:
        with self._lock:
            return bool(self._loop and self._loop.is_running())

    def get_loop(self) -> Optional[asyncio.AbstractEventLoop]:
        with self._lock:
            return self._loop


# 模块级单例：直接 import 使用
bg_loop = BackgroundLoop()
